#ifndef VOIP_CHANNEL_H_
#define VOIP_CHANNEL_H_

#include "ct_noncopyable.h"
#include "ct_udp_socket.h"
#include "ct_udp_msg_buf.h"
#include "ctRWLock.h"
#include "AudioCommand.h"
#include <ext/hash_map>
#include <set>
#include "ct_log_macro.h"

#define NOT_PACK_UDP 1

#define MAX_PACK_UDP_SIZE 640  //UDP数据包合包最大长度
#define MAX_ONE_UDP_CMD_SIZE 10240 //单条udp数据最大长度

//client address records
//map<userid, user addr>

enum {  // media data disable state definition.
        DISABLE_NONE = 0,
        DISABLE_VIDEO_DATA = 0x01,
        DISABLE_DESKTOP_SHARE = 0x02,
        DISABLE_AUDIO_DATA = 0x04,
};

struct peer_info_t {
        struct sockaddr_in addr;
        uint8_t data_flags;   // ref the above enum.
};

typedef __gnu_cxx::hash_map<DWORD, struct peer_info_t> ClientsAddr;
typedef ClientsAddr::iterator CA_IT;

// user set
// set<userid>
typedef std::set<DWORD> UserSet;

class Channel:
        private ctNoncopyable
{
        public:
                Channel(const DWORD &channelid)
                {
                        channel_ = channelid;
                }
                ~Channel()
                {
                        Zebra::logger->info(__PRETTY_FUNCTION__);
                        snd_mutex_.lock();
                        snd_queue_.reset();
                        snd_mutex_.unlock();

                        client_rwlock_.wrlock();
                        clients_.clear();
                        client_rwlock_.unlock();

                        speak_users_.clear();
                        wait_add_users_.clear();
                }
                inline DWORD getChannelID()
                {
                        return channel_;
                }
                inline void addWaitUser(const DWORD &userid)
                {
                        wait_add_users_.insert(userid);
                }
                inline void delWaitUser(const DWORD &userid)
                {
                        wait_add_users_.erase(userid);
                }
                inline bool isWaitUser(const DWORD &userid)
                {
                        return wait_add_users_.find(userid) != wait_add_users_.end();
                }
                inline void addSpeakUser(const DWORD &userid)
                {
                        speak_users_.insert(userid);
                }
                inline void delSpeakUser(const DWORD &userid)
                {
                        speak_users_.erase(userid);
                }
                inline bool isSpeakUser(const DWORD &userid) 
                {
                        return speak_users_.find(userid) != speak_users_.end();
                }
                bool addUser(const DWORD &userid, struct sockaddr_in &addr)
                {
                        if (false == isWaitUser(userid)) {
                                CT_INFO("user(%u) is not in wait list", userid);
                                return false;
                        }
                        client_rwlock_.wrlock();
                        struct peer_info_t cli = { { 0 }, 0 };
                        cli.addr = addr;
                        clients_[userid] = cli;
                        CT_DEBUG("room(%u) user num (%ld)", channel_, clients_.size());
                        client_rwlock_.unlock();
                        return true;
                }
                bool updateUserAddr(const DWORD &userid, struct sockaddr_in &addr)
                {
                        client_rwlock_.wrlock();
                        CA_IT it = clients_.find(userid);
                        if (clients_.end() != it) {
                                it->second.addr = addr;
                                //clients_[userid] = addr;
                                //Zebra::logger->debug(__PRETTY_FUNCTION__);
                        }
                        client_rwlock_.unlock();
                        return true;
                }
                void disableMediaData(uint32_t room_id, uint32_t user_id, uint8_t type, bool enable)
                {
                        CT_TRACE("room(%u) user %d change media data flag(%u, %d)", room_id, user_id,
                                type, enable);

                        ctRWLock_scope_wrlock lock(client_rwlock_);
                        CA_IT it = clients_.find(user_id);
                        if (clients_.end() != it) {
                                if (eUDPDataType_video == type) {
                                        if (enable)
                                                it->second.data_flags &= ~DISABLE_VIDEO_DATA;
                                        else
                                                it->second.data_flags |= DISABLE_VIDEO_DATA;
                                }

                                if (eUDPDataType_desktop == type) {
                                        if (enable)
                                                it->second.data_flags &= ~DISABLE_DESKTOP_SHARE;
                                        else
                                                it->second.data_flags |= DISABLE_DESKTOP_SHARE;
                                }

                                if (eUDPDataType_audio == type) {
                                        if (enable)
                                                it->second.data_flags &= ~DISABLE_AUDIO_DATA;
                                        else
                                                it->second.data_flags |= DISABLE_AUDIO_DATA;
                                }
                                CT_TRACE("#### room(%u) user %u date flags: %02x", room_id, user_id, it->second.data_flags);
                        }
                        else {
                                CT_DEBUG("#### can't find user(%u) in room(%u) for disable media data", user_id, room_id);
                        }
                }

                //在网络层删除用户
                void delUser(const DWORD &userid)
                {
                        client_rwlock_.wrlock();
                        clients_.erase(userid);
                        client_rwlock_.unlock();
                }
                //从应用层删除用户
                void delUser_app(const DWORD &userid)
                {
                        delSpeakUser(userid);
                        delWaitUser(userid);
                        client_rwlock_.wrlock();
                        clients_.erase(userid);
                        client_rwlock_.unlock();
                }
                inline DWORD getUserNum()
                {
                        DWORD num = 0;
                        client_rwlock_.rdlock();
                        num = clients_.size();
                        client_rwlock_.unlock();
                        return num;
                }
                void saveData(const BYTE *data, const DWORD &size)
                {
                        snd_mutex_.lock();
                        snd_queue_.put(data, size);
                        snd_mutex_.unlock();
                }
                void sendData(ctUDPSocket *psocket)
                {
#ifdef NOT_PACK_UDP
                        sendData_nopack(psocket);
#else  //NOT_PACK_UDP
                        sendData_pack(psocket);
#endif //NOT_PACK_UDP
                }
                void sendData_nopack(ctUDPSocket *psocket)
                {
                        if (NULL == psocket) {
                                return ;
                        }
                        snd_mutex_.lock();
                        
                        while (snd_queue_.rd_size()) {
                                if (snd_queue_.rd_size() <= UDP_PKG_LEN) {
                                        Zebra::logger->error("[engine]%s:send buffer data error.sndqueuerdsize:%u", __PRETTY_FUNCTION__, snd_queue_.rd_size());
                                        snd_queue_.reset();
                                } else {
                                        DWORD *tmp = reinterpret_cast<DWORD *>(snd_queue_.rd_buf());
                                        DWORD datalen = *tmp + ctUDPSocket::PH_LEN;
                                        AudioCmd::stUDPPkgHead *cmd = reinterpret_cast<AudioCmd::stUDPPkgHead *>(snd_queue_.rd_buf()+ctUDPSocket::PH_LEN);
                                        if (datalen > snd_queue_.rd_size()) {
                                                Zebra::logger->error("[engine]%s:pkg size bigger than buffer size.cmd(type:%u, roomid:%u, size:%u)", 
                                                                __PRETTY_FUNCTION__, cmd->type_, cmd->room_id_, datalen-ctUDPSocket::PH_LEN);
                                                snd_queue_.reset();
                                        }
                                        if (cmd->type_ == AudioCmd::CMD_AUDIO_FORWARD_NEW) {
                                                AudioCmd::stUDPForwardRoomNew *single = reinterpret_cast<AudioCmd::stUDPForwardRoomNew *>(cmd);
                                                if (0 != single->user_to_) {
                                                        sendToUser(psocket, single->user_to_, snd_queue_.rd_buf(), datalen);
                                                }
                                                else {
                                                        sendToAll(psocket, snd_queue_.rd_buf(), datalen);
                                                }
                                        }
                                        else {
                                                sendToAll(psocket, snd_queue_.rd_buf(), datalen);
                                        }
                                        snd_queue_.rd_flip(datalen);
                                }
                        }
                        snd_mutex_.unlock();
                }
                void sendData_pack(ctUDPSocket *psocket)
                {
                        if (NULL == psocket) {
                                return ;
                        }
                        DWORD mixlen = 0;
                        DWORD freelen= MAX_PACK_UDP_SIZE;
                        BYTE mixdata[MAX_PACK_UDP_SIZE] = {0}; //合包内容

                        DWORD *tmp = NULL;
                        DWORD datalen = 0;
                        snd_mutex_.lock();
                        while (snd_queue_.rd_size()) {
                                if (snd_queue_.rd_size() <= UDP_PKG_LEN) {
                                        Zebra::logger->error("[engine]%s:send buffer data error.sndqueuerdsize:%u", __FUNCTION__, snd_queue_.rd_size());
                                        snd_queue_.reset();
                                }
                                else {
                                        tmp = reinterpret_cast<DWORD *>(snd_queue_.rd_buf());
                                        datalen = *tmp + ctUDPSocket::PH_LEN;
                                        if (datalen > freelen) {
                                                if (mixlen != 0) { //把之前的发送掉
                                                        sendToAll(psocket, mixdata, mixlen);
                                                        mixlen = 0;
                                                        freelen = MAX_PACK_UDP_SIZE;
                                                        bzero(mixdata, MAX_PACK_UDP_SIZE);
                                                }
                                        }
                                        if (datalen <= freelen) {
                                                AudioCmd::stUDPPkgHead *cmd = reinterpret_cast<AudioCmd::stUDPPkgHead *>(snd_queue_.rd_buf()+ctUDPSocket::PH_LEN);
                                                if (cmd->type_ == AudioCmd::CMD_AUDIO_FORWARD_NEW) {
                                                        AudioCmd::stUDPForwardRoomNew *single = reinterpret_cast<AudioCmd::stUDPForwardRoomNew *>(cmd);
                                                        if (0 != single->user_to_) {
                                                                sendToUser(psocket, single->user_to_, snd_queue_.rd_buf(), datalen);
                                                        }
                                                        else {
                                                                bcopy(tmp, mixdata+mixlen, datalen);
                                                                freelen -= datalen;
                                                                mixlen  += datalen;
                                                        }
                                                }
                                                else {
                                                        bcopy(tmp, mixdata+mixlen, datalen);
                                                        freelen -= datalen;
                                                        mixlen  += datalen;
                                                }
                                        }
                                        else if (datalen < MAX_ONE_UDP_CMD_SIZE) { //针对过大的指令的处理
                                                if (mixlen != 0) { //把之前的发送掉
                                                        sendToAll(psocket, mixdata, mixlen);
                                                        mixlen = 0;
                                                        freelen = MAX_PACK_UDP_SIZE;
                                                        bzero(mixdata, MAX_PACK_UDP_SIZE);
                                                }

                                                AudioCmd::stUDPPkgHead *cmd = reinterpret_cast<AudioCmd::stUDPPkgHead *>(snd_queue_.rd_buf()+ctUDPSocket::PH_LEN);
                                                if (cmd->type_ == AudioCmd::CMD_AUDIO_FORWARD_NEW) {
                                                        AudioCmd::stUDPForwardRoomNew *single = reinterpret_cast<AudioCmd::stUDPForwardRoomNew *>(cmd);
                                                        if (0 != single->user_to_) {
                                                                sendToUser(psocket, single->user_to_, snd_queue_.rd_buf(), datalen);
                                                        }
                                                        else {
                                                                sendToAll(psocket, snd_queue_.rd_buf(), datalen);
                                                        }
                                                }
                                                else {
                                                        sendToAll(psocket, snd_queue_.rd_buf(), datalen);
                                                }
                                        }
                                        snd_queue_.rd_flip(datalen);
                                }
                        }
                        snd_mutex_.unlock();

                        if (mixlen != 0) { //把之前的发送掉
                                sendToAll(psocket, mixdata, mixlen);
                        }
                }


        private:
                DWORD channel_;

                ClientsAddr clients_;
                ctRWLock client_rwlock_;

                UserSet speak_users_;
                UserSet wait_add_users_;

                t_BufferCmdQueue snd_queue_;	//send buffer queue
                ctMutex snd_mutex_;

                void sendToAll(ctUDPSocket *psocket, BYTE *data, const DWORD &datalen)
                {
                        if (NULL == psocket) {
                                return ;
                        }
                        client_rwlock_.rdlock();
                        uint8_t cmd_type = *(uint8_t*)(data + sizeof(uint32_t));
                        uint8_t data_type = 0;
                        uint32_t user_id;
                        // try to get data type
                        if (AudioCmd::CMD_AUDIO_FORWARD == cmd_type) {
                                AudioCmd::stUDPForwardRoom* hdr = (AudioCmd::stUDPForwardRoom*)(data + sizeof(uint32_t));
                                data_type = hdr->data_type_;
                                user_id = hdr->user_id_;
                        } else if (AudioCmd::CMD_AUDIO_FORWARD_NEW == cmd_type) {
                                AudioCmd::stUDPForwardRoomNew* hdr = (AudioCmd::stUDPForwardRoomNew*)(data + sizeof(uint32_t));
                                data_type = hdr->data_type_;
                                user_id = hdr->user_from_;
                        } else if (AudioCmd::CMD_AUDIO_FORWARD_NEW_NEW == cmd_type) {
                                AudioCmd::stUDPForwardRoomNewNew* hdr = (AudioCmd::stUDPForwardRoomNewNew*)(data + sizeof(uint32_t));
                                data_type = hdr->data_type_;
                                user_id = hdr->user_from_;
                        }

                        //CT_TRACE("@@@@media type(%u) from user(%u): %u", cmd_type, user_id, data_type);
                        for(CA_IT it = clients_.begin(); it != clients_.end(); it++) {
                                const peer_info_t& peer = it->second;

                                if ((eUDPDataType_audio == data_type && (peer.data_flags & DISABLE_AUDIO_DATA)) ||
                                    (eUDPDataType_video == data_type && (peer.data_flags & DISABLE_AUDIO_DATA)) ||
                                    (eUDPDataType_desktop == data_type && (peer.data_flags & DISABLE_DESKTOP_SHARE)))
                                        continue;
                                
                                int ret = psocket->sendData(&(it->second.addr), data, datalen);
                                if (-1 == ret) {
                                        CT_WARN("[engine]%s:send data error", __PRETTY_FUNCTION__);
                                }
                                else if ((DWORD)ret != datalen && 0 != ret) {
                                        CT_TRACE("[engine]%s:send data size:%u", __PRETTY_FUNCTION__, ret);
                                }
                                /*
                                   else Zebra::logger->debug("[engine][channel mix]send to user(%u) len(%u)", it->first, mixlen);
                                   */
                        }
                        client_rwlock_.unlock();
                }
                void sendToUser(ctUDPSocket *psocket, const DWORD &userid, BYTE *data,  const DWORD &datalen)
                {
                        //Zebra::logger->debug("----send data to user(%u)", userid);
                        if (NULL == psocket) {
                                return ;
                        }
                        client_rwlock_.rdlock();
                        CA_IT it = clients_.find(userid);
                        if (clients_.end() != it) {
                                int ret = psocket->sendData(&(it->second.addr), data, datalen);
                                if (-1 == ret) {
                                        CT_WARN("[engine]%s:send data error", __PRETTY_FUNCTION__);
                                }
                                else if ((DWORD)ret != datalen 
                                                && 0 != ret) {
                                        CT_TRACE("[engine]%s:send data size:%u", __PRETTY_FUNCTION__, ret);
                                }
                        }
                        else {
                                //Zebra::logger->debug("----user(%u) not exist", userid);
                        }
                        client_rwlock_.unlock();
                }
};

#endif// VOIP_CHANNEL_H_
